--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit e46eed7f7e2637c4c21f4c772d908ef75ec49e6a
Parents : 5760566
Author : Ivan <ivan@quad4.io>
Signature : Invalid signer <e46112d44649266d71fe2193e00a4710>, author is <ivan@quad4.io>
Date : 2026-06-19T14:50:13-05:00
feat(rnode): implement TX power validation and normalization for RNode interfaces, enhancing configuration integrity and preventing startup crashes
Changes
7 files changed, 272 insertions(+), 7 deletions(-)
Diff
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index bff9a743..ab9e1d2d 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -897,10 +897,12 @@ class ReticulumMeshChat:
except Exception:
pass
from meshchatx.src.backend.rnode_support import (
+ guard_invalid_rnode_txpower_in_config,
guard_rnode_interfaces_on_android,
)
guard_rnode_interfaces_on_android(config_path)
+ guard_invalid_rnode_txpower_in_config(config_path)
def setup_identity(self, identity: RNS.Identity):
identity_hash = identity.hash.hex()
@@ -5550,12 +5552,15 @@ class ReticulumMeshChat:
status=422,
)
- # ensure txpower provided
+ # ensure txpower provided and within Reticulum limits
interface_txpower = data.get("txpower")
- if interface_txpower is None or interface_txpower == "":
+ txpower_error = InterfaceEditor.validate_rnode_txpower(
+ interface_txpower,
+ )
+ if txpower_error is not None:
return web.json_response(
{
- "message": "TX power is required",
+ "message": txpower_error,
},
status=422,
)
@@ -5588,7 +5593,9 @@ class ReticulumMeshChat:
)
)
interface_details["bandwidth"] = interface_bandwidth
- interface_details["txpower"] = interface_txpower
+ interface_details["txpower"] = InterfaceEditor.normalize_rnode_txpower(
+ interface_txpower,
+ )
interface_details["spreadingfactor"] = interface_spreadingfactor
interface_details["codingrate"] = interface_codingrate
@@ -5672,6 +5679,17 @@ class ReticulumMeshChat:
status=422,
)
+ sub_txpower_error = InterfaceEditor.validate_rnode_txpower(
+ sub_interface.get("txpower"),
+ )
+ if sub_txpower_error is not None:
+ return web.json_response(
+ {
+ "message": f"Sub-interface {idx + 1}: {sub_txpower_error}",
+ },
+ status=422,
+ )
+
sub_interface_name = sub_interface.get("name")
interface_details[sub_interface_name] = {
"interface_enabled": "true",
@@ -5679,7 +5697,9 @@ class ReticulumMeshChat:
sub_interface["frequency"],
),
"bandwidth": int(sub_interface["bandwidth"]),
- "txpower": int(sub_interface["txpower"]),
+ "txpower": InterfaceEditor.normalize_rnode_txpower(
+ sub_interface["txpower"],
+ ),
"spreadingfactor": int(sub_interface["spreadingfactor"]),
"codingrate": int(sub_interface["codingrate"]),
"vport": int(sub_interface["vport"]),
@@ -5995,14 +6015,54 @@ class ReticulumMeshChat:
iface_body["frequency"] = (
InterfaceEditor.coerce_rnode_frequency_hz(freq)
)
+ txpower = iface_body.get("txpower")
+ if txpower is not None and txpower != "":
+ txpower_error = InterfaceEditor.validate_rnode_txpower(
+ txpower,
+ )
+ if txpower_error is not None:
+ return web.json_response(
+ {
+ "message": (
+ f'Interface "{interface_name}": {txpower_error}'
+ ),
+ },
+ status=422,
+ )
+ iface_body["txpower"] = (
+ InterfaceEditor.normalize_rnode_txpower(txpower)
+ )
elif iface_type == "RNodeMultiInterface":
- for _sub_key, sub in list(iface_body.items()):
+ for sub_key, sub in list(iface_body.items()):
if isinstance(sub, dict):
freq = sub.get("frequency")
if freq is not None and freq != "":
sub["frequency"] = (
InterfaceEditor.coerce_rnode_frequency_hz(freq)
)
+ txpower = sub.get("txpower")
+ if txpower is not None and txpower != "":
+ txpower_error = (
+ InterfaceEditor.validate_rnode_txpower(
+ txpower,
+ )
+ )
+ if txpower_error is not None:
+ return web.json_response(
+ {
+ "message": (
+ f'Interface "{interface_name}" '
+ f'sub-interface "{sub_key}": '
+ f"{txpower_error}"
+ ),
+ },
+ status=422,
+ )
+ sub["txpower"] = (
+ InterfaceEditor.normalize_rnode_txpower(
+ txpower,
+ )
+ )
# update reticulum config with new interfaces
interfaces = self._get_interfaces_section()
diff --git a/meshchatx/src/backend/interface_editor.py b/meshchatx/src/backend/interface_editor.py
index a43d51e2..814beff5 100644
--- a/meshchatx/src/backend/interface_editor.py
+++ b/meshchatx/src/backend/interface_editor.py
@@ -71,9 +71,38 @@ def coerce_rnode_frequency_hz(value):
return int(round(f))
+RNODE_TXPOWER_MIN = 0
+RNODE_TXPOWER_MAX = 37
+
+
+def normalize_rnode_txpower(value):
+ """Return integer dBm for Reticulum ``RNodeInterface`` config."""
+ if value is None or value == "":
+ return value
+ return int(float(str(value).strip()))
+
+
+def validate_rnode_txpower(value) -> str | None:
+ """Return an API error message when TX power is invalid for Reticulum."""
+ if value is None or value == "":
+ return "TX power is required"
+ try:
+ power = normalize_rnode_txpower(value)
+ except (TypeError, ValueError):
+ return "TX power must be an integer dBm value"
+ if power < RNODE_TXPOWER_MIN or power > RNODE_TXPOWER_MAX:
+ return (
+ f"TX power must be between {RNODE_TXPOWER_MIN} and {RNODE_TXPOWER_MAX} dBm "
+ "(Reticulum RNodeInterface limit; typical SX1262 range is 0-22 dBm)"
+ )
+ return None
+
+
class InterfaceEditor:
coerce_rnode_frequency_hz = staticmethod(coerce_rnode_frequency_hz)
normalize_rnode_tcp_port = staticmethod(normalize_rnode_tcp_port)
+ normalize_rnode_txpower = staticmethod(normalize_rnode_txpower)
+ validate_rnode_txpower = staticmethod(validate_rnode_txpower)
@staticmethod
def minimum_fixed_mtu() -> int:
diff --git a/meshchatx/src/backend/rnode_support.py b/meshchatx/src/backend/rnode_support.py
index dfffb8fa..e2183246 100644
--- a/meshchatx/src/backend/rnode_support.py
+++ b/meshchatx/src/backend/rnode_support.py
@@ -85,6 +85,75 @@ def disable_rnode_interfaces_in_config(config_path: str) -> bool:
return modified
+def _rnode_interface_has_invalid_txpower(iface: dict) -> bool:
+ from meshchatx.src.backend.interface_editor import validate_rnode_txpower
+
+ iface_type = iface.get("type", "")
+ if not isinstance(iface_type, str):
+ return False
+ if iface_type in ("RNodeInterface", "RNodeIPInterface"):
+ return validate_rnode_txpower(iface.get("txpower")) is not None
+ if iface_type == "RNodeMultiInterface":
+ for value in iface.values():
+ if isinstance(value, dict) and "txpower" in value:
+ if validate_rnode_txpower(value.get("txpower")) is not None:
+ return True
+ return False
+
+
+def guard_invalid_rnode_txpower_in_config(config_path: str) -> bool:
+ """Disable RNode interfaces whose TX power would crash Reticulum on startup."""
+ import os
+
+ if not os.path.isfile(config_path):
+ return False
+ try:
+ from RNS.vendor.configobj import ConfigObj
+
+ from meshchatx.src.backend.interface_editor import validate_rnode_txpower
+
+ cfg = ConfigObj(config_path)
+ except Exception:
+ return False
+
+ modified = False
+ interfaces = cfg.get("interfaces")
+ if not isinstance(interfaces, dict):
+ return False
+ for iface_name, iface in interfaces.items():
+ if not isinstance(iface, dict):
+ continue
+ if not _rnode_interface_has_invalid_txpower(iface):
+ continue
+ if str(iface.get("interface_enabled", "")).lower() not in (
+ "true",
+ "yes",
+ "1",
+ "on",
+ ):
+ continue
+ iface["interface_enabled"] = "false"
+ modified = True
+ txpower = iface.get("txpower")
+ if txpower is None:
+ for value in iface.values():
+ if isinstance(value, dict) and "txpower" in value:
+ txpower = value.get("txpower")
+ break
+ detail = validate_rnode_txpower(txpower) or "invalid TX power"
+ logger.warning(
+ 'Disabled RNode interface "%s" before startup: %s',
+ iface_name,
+ detail,
+ )
+ if modified:
+ try:
+ cfg.write()
+ except Exception:
+ pass
+ return modified
+
+
def guard_rnode_interfaces_on_android(config_path: str) -> bool:
"""On Android without usbserial4a, disable RNode interfaces to avoid startup crashes."""
if not _is_chaquopy_android():
diff --git a/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue b/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue
index 38e83231..7fa3ddea 100644
--- a/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue
+++ b/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue
@@ -801,8 +801,11 @@
<div>
<FormLabel class="glass-label">Power (dBm)</FormLabel>
<input
- v-model="newInterfaceTxpower"
+ v-model.number="newInterfaceTxpower"
type="number"
+ :min="RNodeInterfaceDefaults.txpowerMin"
+ :max="RNodeInterfaceDefaults.txpowerMax"
+ step="1"
class="input-field"
/>
</div>
@@ -2027,6 +2030,8 @@ export default {
8, // 4:8
],
spreadingfactors: [5, 6, 7, 8, 9, 10, 11, 12],
+ txpowerMin: 0,
+ txpowerMax: 37,
},
RNodeInterfaceLoRaParameters: {
diff --git a/tests/backend/test_interface_editor.py b/tests/backend/test_interface_editor.py
index c5701d02..ab1ee533 100644
--- a/tests/backend/test_interface_editor.py
+++ b/tests/backend/test_interface_editor.py
@@ -46,6 +46,25 @@ def test_coerce_rnode_frequency_hz_leaves_midrange_hz():
assert InterfaceEditor.coerce_rnode_frequency_hz(125000) == 125000
+def test_normalize_rnode_txpower_integer():
+ assert InterfaceEditor.normalize_rnode_txpower(7) == 7
+ assert InterfaceEditor.normalize_rnode_txpower("14") == 14
+ assert InterfaceEditor.normalize_rnode_txpower("7.0") == 7
+
+
+def test_validate_rnode_txpower_accepts_reticulum_range():
+ assert InterfaceEditor.validate_rnode_txpower(0) is None
+ assert InterfaceEditor.validate_rnode_txpower(22) is None
+ assert InterfaceEditor.validate_rnode_txpower(37) is None
+
+
+def test_validate_rnode_txpower_rejects_out_of_range():
+ assert InterfaceEditor.validate_rnode_txpower(-9) is not None
+ assert InterfaceEditor.validate_rnode_txpower(38) is not None
+ assert InterfaceEditor.validate_rnode_txpower("bad") is not None
+ assert InterfaceEditor.validate_rnode_txpower(None) is not None
+
+
def test_normalize_rnode_tcp_port_host_only():
assert (
InterfaceEditor.normalize_rnode_tcp_port("tcp://10.0.0.5") == "tcp://10.0.0.5"
diff --git a/tests/backend/test_interface_options.py b/tests/backend/test_interface_options.py
index 27667f33..4c52acac 100644
--- a/tests/backend/test_interface_options.py
+++ b/tests/backend/test_interface_options.py
@@ -533,6 +533,49 @@ async def test_rnode_frequency_mhz_decimal_normalized_to_hz(temp_dir):
assert config["interfaces"]["RadioEU"]["frequency"] == 868825000
+@pytest.mark.asyncio
+async def test_rnode_rejects_invalid_txpower(temp_dir):
+ config = ConfigDict({"reticulum": {}, "interfaces": {}})
+
+ async with make_app(temp_dir, config) as handler:
+ payload = {
+ "name": "RadioBad",
+ "type": "RNodeInterface",
+ "port": "/dev/ttyUSB0",
+ "frequency": 868000000,
+ "bandwidth": 125000,
+ "txpower": -9,
+ "spreadingfactor": 8,
+ "codingrate": 5,
+ }
+ response = await handler(make_request(payload))
+ body = json.loads(response.body)
+ assert response.status == 422, body
+ assert "TX power" in body["message"]
+ assert "RadioBad" not in config["interfaces"]
+
+
+@pytest.mark.asyncio
+async def test_rnode_normalizes_txpower_to_integer(temp_dir):
+ config = ConfigDict({"reticulum": {}, "interfaces": {}})
+
+ async with make_app(temp_dir, config) as handler:
+ payload = {
+ "name": "Radio",
+ "type": "RNodeInterface",
+ "port": "/dev/ttyUSB0",
+ "frequency": 868000000,
+ "bandwidth": 125000,
+ "txpower": "14.0",
+ "spreadingfactor": 8,
+ "codingrate": 5,
+ }
+ response = await handler(make_request(payload))
+ body = json.loads(response.body)
+ assert response.status == 200, body
+ assert config["interfaces"]["Radio"]["txpower"] == 14
+
+
@pytest.mark.asyncio
async def test_kiss_persists_full_serial_options(temp_dir):
config = ConfigDict({"reticulum": {}, "interfaces": {}})
diff --git a/tests/backend/test_rnode_support.py b/tests/backend/test_rnode_support.py
index f54bad2d..fd0a242b 100644
--- a/tests/backend/test_rnode_support.py
+++ b/tests/backend/test_rnode_support.py
@@ -42,3 +42,43 @@ def test_guard_keeps_rnode_when_usbserial4a_available(tmp_path, monkeypatch):
def test_rnode_serial_supported_on_desktop(monkeypatch):
monkeypatch.setattr(rnode_support, "_is_chaquopy_android", lambda: False)
assert rnode_support.rnode_serial_supported() is True
+
+
+def test_guard_disables_rnode_with_invalid_txpower(tmp_path):
+ config_path = tmp_path / "config"
+ config_path.write_text(
+ """[interfaces]
+[[t9-srv]]
+type = RNodeInterface
+interface_enabled = True
+port = /dev/ttyUSB0
+frequency = 868000000
+bandwidth = 125000
+txpower = -9
+spreadingfactor = 8
+codingrate = 5
+""",
+ encoding="utf-8",
+ )
+
+ assert rnode_support.guard_invalid_rnode_txpower_in_config(str(config_path)) is True
+ text = config_path.read_text(encoding="utf-8")
+ assert "interface_enabled = false" in text.lower()
+
+
+def test_guard_keeps_rnode_with_valid_txpower(tmp_path):
+ config_path = tmp_path / "config"
+ config_path.write_text(
+ """[interfaces]
+[[Radio]]
+type = RNodeInterface
+interface_enabled = True
+txpower = 7
+""",
+ encoding="utf-8",
+ )
+
+ assert (
+ rnode_support.guard_invalid_rnode_txpower_in_config(str(config_path)) is False
+ )
+ assert "interface_enabled = True" in config_path.read_text(encoding="utf-8")
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────